package org.arch_learn.rpc.nz_rpc.configuration;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.arch_learn.rpc.nz_rpc.endpoint.NettyRpcClient;
import org.arch_learn.rpc.nz_rpc.handler.NettyRpcClientHandler;
import org.arch_learn.rpc.nz_rpc.properties.NzRpcProperties;
import org.arch_learn.rpc.nz_rpc.utils.ScanUtils;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

@Configuration
@ConditionalOnBean(NzRpcInitializationConfiguration.Marker.class)
@EnableConfigurationProperties(value = {NzRpcProperties.class})
public class RPCAutoConfiguration {

    @Bean
    public CuratorFramework curatorFramework(NzRpcProperties nzRpcProperties) {
        System.out.println("======================================");
        ExponentialBackoffRetry exponentialBackoffRetry = new ExponentialBackoffRetry(1000, 3);
        String zookeeper = nzRpcProperties.getZookeeper();
        //使用fluent编程风格
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString(zookeeper)
                .sessionTimeoutMs(5000)
                .connectionTimeoutMs(3000)
                .retryPolicy(exponentialBackoffRetry)
                //设置了一个独立的命名空间，有利于实现不同zookeeper业务的隔离
                .namespace("nz_rpc")
                .build();
        client.start();
        System.out.println("zookeeper被创建了。。。。");
        return client;
    }

    @Bean(name = "nettyRpcClients")
    public ConcurrentHashMap<String, List<NettyRpcClient>>
    nettyRpcClients(CuratorFramework client,
                    NzRpcProperties nzRpcProperties) throws Exception {

        String rpcServicesPackage = nzRpcProperties.getServicePackage();
        String rpcReferencePackage = nzRpcProperties.getReferencePackage();
        Set<String> rpcServices = new HashSet<>();
        Set<String> rpcReferences = new HashSet<>();
        if (null != rpcServicesPackage) {
            ScanUtils.loadServices(rpcServicesPackage, rpcServices);
        }

        if (null != rpcReferencePackage) {
            ScanUtils.loadReferences(rpcReferencePackage, rpcReferences);
        }
        String netty = nzRpcProperties.getNetty();
        //如果提供服务则需要注册到zookeeper上（添加相应节点）
        for (String service : rpcServices) {
            String childPath = "/" + service + "/" + netty;
            client.create()
                    .creatingParentsIfNeeded()
                    .withMode(CreateMode.EPHEMERAL)
                    .forPath(childPath, "0 0".getBytes());

        }
        ConcurrentHashMap<String, List<NettyRpcClient>> servicesMap = new ConcurrentHashMap<>();

        //通过获取引用的服务，来实例化netty客户端
        for (String service : rpcReferences) {
            List<String> children = client.getChildren().forPath("/" + service);
            List<NettyRpcClient> nettyRpcServers = children.stream().map(c -> {
                NettyRpcClient nettyRpcClient = new NettyRpcClient(c);
                nettyRpcClient.initClient(new NettyRpcClientHandler());
                return nettyRpcClient;
            }).collect(Collectors.toList());
            List<NettyRpcClient> nettyRpcEndpoints1 = servicesMap.get(service);
            if (null == nettyRpcEndpoints1 || nettyRpcEndpoints1.size() == 0) {
                nettyRpcEndpoints1 = new ArrayList<>();
                nettyRpcEndpoints1.addAll(nettyRpcServers);
                servicesMap.put(service, nettyRpcEndpoints1);
            }
        }

        //给所有引用服务的zookeeper节点注册子节点变更监听器
        rpcReferences.forEach(r -> {
            PathChildrenCache childrenCache = new PathChildrenCache(client, "/" + r, true);
            try {
                /**
                 * StartMode:初始化方式
                 *	POST_INITIALIZED_EVENT: 异步初始化，初始化之后会触发事件(会进入下面的第一个case)
                 *	NORMAL:异步初始化 (不会进入下面的第一个case)
                 *	BUILD_INITIAL_CACHE: 同步初始化(把节点数据同步缓存到Cache中)
                 */
                childrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
                childrenCache.getListenable().addListener(pathChildrenCacheListener(servicesMap));
            } catch (Exception e) {
                e.printStackTrace();
            }
        });

        return servicesMap;
    }


    //服务子节点（服务提供实例ip:port）监听器生成方法
    public PathChildrenCacheListener pathChildrenCacheListener(ConcurrentHashMap<String, List<NettyRpcClient>> servicesMap) {
        return new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent pathChildrenCacheEvent) throws Exception {
                PathChildrenCacheEvent.Type type = pathChildrenCacheEvent.getType();
                if (pathChildrenCacheEvent.getData() == null) return;
                String childPath = pathChildrenCacheEvent.getData().getPath();
                String[] arr = childPath.split("/");
                String serviceName = arr[1];
                List<NettyRpcClient> nettyRpcClients = servicesMap.get(serviceName);
                switch (type) {
                    case CHILD_ADDED:
                        NettyRpcClient nettyRpcClient = new NettyRpcClient(arr[2]);
                        nettyRpcClient.initClient(new NettyRpcClientHandler());
                        nettyRpcClients.add(nettyRpcClient);
                        removeRedundant(serviceName, nettyRpcClients, servicesMap);
                        System.out.println("服务上线线后还剩：");
                        servicesMap.get(serviceName).forEach(v -> System.out.println(v.toString()));
                        break;
                    case CHILD_REMOVED:
                        nettyRpcClients.removeIf(v -> v.getConnectStr().equals(arr[2]));
                        removeRedundant(serviceName, nettyRpcClients, servicesMap);
                        System.out.println("服务下线后还剩：");
                        servicesMap.get(serviceName).forEach(v -> System.out.println(v.toString()));
                        break;
                    default:
                        break;
                }
            }
        };
    }

    //发现有时候会出现重复的客户端实例，这里做去重处理
    private void removeRedundant(String serviceName, List<NettyRpcClient> nettyRpcClients, ConcurrentHashMap<String, List<NettyRpcClient>> servicesMap) {
        List<NettyRpcClient> leftClients = nettyRpcClients.stream()
                .collect(Collectors.collectingAndThen(
                        Collectors.toCollection(() -> new TreeSet<>(Comparator.comparing(
                                NettyRpcClient::getConnectStr
                        ))), ArrayList::new
                ));
        nettyRpcClients.removeAll(leftClients);
//        nettyRpcClients.forEach(v -> v.close());
        servicesMap.put(serviceName, leftClients);
    }
}
